package sparkStreaming.taxi_hailingCount

import java.time.LocalDateTime
import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import scala.collection.mutable
import scala.util.Random

/**
 * @Author:lixinlei
 * @Date:2021/6/3 15:21
 **/
object KafkaStreamingProducer {

  //初始化生成测试数据所需原始数据
  //号段数组
  val phone_array = Array(130,159,181,133,177)
  //城市列表
  val city_array = Array("哈尔滨","齐齐哈尔","牡丹江")

  //各城市区域列表
  val area_hrb = Array("南岗区","道里区","道外区","香坊区","松北区","平房区","呼兰区","阿城区")

  val area_qqhe = Array("龙沙区","建华区","龙江县")

  val area_mdj = Array("爱民区","东安区","西安区","阳明区")

  val rand = new Random()


  /**
   * 生成测试数据
   * phonenum city+area time distance price
   */
  def genTestData() = {

    val dataList = mutable.ListBuffer[String]()

    //调用一次此方法，生成10条数据
    for (i <- 1 to 10) {
      //生成第一个字段-手机号
      val phonenum = new StringBuffer()
      phonenum.append(phone_array(rand.nextInt(phone_array.length)))
      for(j <- 1 to 8){
        phonenum.append(rand.nextInt(10))
      }

      //生成第二个字段-城市+区域
      val city = city_array(rand.nextInt(city_array.length))
      var area = ""
      city match {
        case "哈尔滨" => {
          area = area_hrb(rand.nextInt(area_hrb.length))
        }
        case "齐齐哈尔" => {
          area = area_qqhe(rand.nextInt(area_qqhe.length))
        }
        case "牡丹江" => {
          area = area_mdj(rand.nextInt(area_mdj.length))
        }
      }
      val address = city+area

      //生成第三个字段-时间
      val dateTime = LocalDateTime.now().toString


      //生成第四个字段-距离
      val distance = rand.nextInt(201)

      //生成第五个字段-价格
      var price = 10
      if(distance>3){
        val morePrice = 2 * (distance-3)
        price += morePrice
      }

//      println("生成==="+phonenum.toString+","+address+","+dateTime+","+distance+","+price)
      dataList.append(phonenum.toString+","+address+","+dateTime+","+distance+","+price)
    }

    dataList
  }

  /**
   * kafka生产者 将随机产生的数据发送到kafka
   */
  def kafkaProducer() = {

    //设置kafka生产者所需属性
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,Common.BOOTSTRAP)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,Common.SERIALIZER_CLASS)
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,Common.SERIALIZER_CLASS)

    //创建kafka生产者对象
    val producer = new KafkaProducer[String,String](prop)

    //无限循环，模拟永不停止的实时数据
    while(true){

      val list = genTestData()
      list.append("13888888888,哈尔滨松北区,2021-06-09T14:17:20.208,20,200")
      list.append("13888888888,哈尔滨松北区,2021-06-09T14:17:20.208,20,200")
      list.append("13888888888,哈尔滨松北区,2021-06-09T14:17:20.208,20,200")
      list.append("13888888888,哈尔滨松北区,2021-06-10T14:17:20.208,20,200")
      list.append("13888888888,哈尔滨松北区,2021-06-10T14:17:20.208,20,200")
      list.append("13888888888,哈尔滨松北区,2021-07-08T14:17:20.208,20,200")
      list.foreach(
        data => {
          //把每条数据发送给kafka的具体某个主题
          val record = new ProducerRecord[String,String](Common.KAFKA_TOPIC_ORACLE,data)
          producer.send(record)
        }
      )

      //每3秒钟生成10条数据(genTestData方法中一次生成10条)
      Thread.sleep(3000)
    }
  }

  def main(args: Array[String]): Unit = {
    kafkaProducer()
  }

}
